package edu.ntu.im.p2ptrade;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import edu.ntu.im.p2ptrade.entity.Item;
import edu.ntu.im.p2ptrade.entity.ItemRepository;
import edu.ntu.im.p2ptrade.entity.Transaction;
import edu.ntu.im.p2ptrade.entity.TransactionRequest;
import edu.ntu.im.p2ptrade.msg.AbortCommit;
import edu.ntu.im.p2ptrade.msg.ApproveTRequest;
import edu.ntu.im.p2ptrade.msg.AuthenticatedMessageWrapper;
import edu.ntu.im.p2ptrade.msg.CanCommit;
import edu.ntu.im.p2ptrade.msg.CannotCommit;
import edu.ntu.im.p2ptrade.msg.Commit;
import edu.ntu.im.p2ptrade.msg.DenyTRequest;
import edu.ntu.im.p2ptrade.msg.OKCommit;
import edu.ntu.im.p2ptrade.msg.PMessage;
import edu.ntu.im.p2ptrade.msg.TRequest;
import edu.ntu.im.p2ptrade.msg.handle.EventRegister;

public class TransactionManager implements LockListener {
	private Map<String, TransactionRequest> incomingReq = new HashMap<String, TransactionRequest>();
	private Map<String, TransactionRequest> issuedReq = new HashMap<String, TransactionRequest>();
	private Map<String, TransactionRequest> deniedReq = new HashMap<String, TransactionRequest>();
	private Map<String, TransactionRequest> abortedReq = new HashMap<String, TransactionRequest>();
	private Map<String, TransactionRequest> successReq = new HashMap<String, TransactionRequest>();

	private Set<String> lockFailedTransId = new HashSet<String>();

	private Set<String> lockSuccessTransId = new HashSet<String>();

	private Object mutex = new Object();

	private static TransactionManager me;
	
	public CommitInfo  ifCommit = null;
	public CommitInfo  commit = null;
	public CommitInfo  abortCommit = null;
	
	public  Coordinator coordinator = null;

	private TransactionManager() {
		EventRegister.lockListener.add(this);

	}

	public void issueTransactionRequest(Transaction.TradeInfo mine,
			Transaction.TradeInfo peers) {
		System.out.println(P2PTrade.getInstance().getPeerName()
				+ ":issue transaction");
		String transactionId = generateTracsactionId(peers.getOwner());
		Transaction trans = new Transaction();
		trans.setPart1(mine);
		trans.setPart2(peers);
		trans.setId(transactionId);
		TransactionRequest tr = new TransactionRequest();
		tr.setTrans(trans);
		PMessage mtr = new AuthenticatedMessageWrapper(new TRequest(tr));

		this.issuedReq.put(transactionId, tr);
		try {
			Util.transdferMessageToPeer(mtr.getMessageToTransfer(), peers
					.getOwner());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static String generateTracsactionId(String peer) {
		String orignal = P2PTrade.getInstance().getPeerName() + "_" + peer
				+ new Date().getTime();
		MessageDigest digest = null;
		try {
			digest = MessageDigest.getInstance("SHA1");
		} catch (NoSuchAlgorithmException e) {
			e.printStackTrace();
		}
		return Util.dumpBytes(digest.digest(orignal.getBytes()));
	}

	public static TransactionManager getInstance() {
		if (me == null) {
			me = new TransactionManager();
		}
		return me;
	}

	public void addTransactionRequest(TransactionRequest tr) {
		this.incomingReq.put(tr.getTrans().getId(), tr);
	}

	public void transactionRequestDenied(String trasactionId) {
		if (!this.deniedReq.containsKey(trasactionId)) {
			TransactionRequest tr = this.issuedReq.remove(trasactionId);
			this.deniedReq.put(trasactionId, tr);
		}
	}
	
	public void askedCanCommit(String fromWhom, String transactionId){
		this.ifCommit = new CommitInfo(transactionId, fromWhom);
	}

	public synchronized void transactionRequestApproved(String peer,
			String transactionId) {
		System.out.println(P2PTrade.getInstance().getPeerName()
				+ ":transactionRequestApproved");
		 coordinator = new Coordinator(transactionId, peer);
		
		new Thread(coordinator).start();
		
		while (ifCommit == null){
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		TransactionRequest tr = this.issuedReq.get(transactionId);
		
		List<Item> items = null;
		if (P2PTrade.getInstance().getPeerName().equals(
				tr.getTrans().getPart1().getOwner())) {
			items = tr.getTrans().getPart1().getItems();
		} else {
			items = tr.getTrans().getPart2().getItems();
		}
		boolean hasThis = true;
		for (Item item:items){
			if (!ItemRepository.getInstance().hasThis(item.getId())){
				hasThis =  false;
			}
		}
		
		boolean lockme = LockManager.getInstance().lockMySelf(transactionId);
		System.out.println(this.ifCommit.fromWhom);
		if (!lockme || !hasThis){
			CannotCommit cannot = new CannotCommit(transactionId);
			Util.transdferMessageToPeer(cannot.getMessageToTransfer(), this.ifCommit.fromWhom);
		}else {
			OKCommit ok = new OKCommit(transactionId);
			Util.transdferMessageToPeer(ok.getMessageToTransfer(), this.ifCommit.fromWhom);
		}
		
		while (this.commit == null && this.abortCommit == null){
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		if (this.commit!=null){
			 tr = this.issuedReq.remove(transactionId);
			
			synchronized (mutex) {
				ItemRepository.getInstance().getItems().removeAll(items);
				for (Item item: items){
					System.out.println(item.toString());
				}
			}
			
			System.out.println(P2PTrade.getInstance().getPeerName()
					+ ":before unlock to peer:" + peer);

			LockManager.getInstance().unlockMySelf(transactionId);
			this.successReq.put(transactionId, tr);
			System.out.println(P2PTrade.getInstance().getPeerName()
					+ ":transaction finished!!");
		}else {
			this.transactionRequestAborted(transactionId);
		}
		this.ifCommit = null;
		this.abortCommit = null;
		this.commit = null;
		this.coordinator = null;
		

	}

	
	public void approveRequest(String transactionId) {
		TransactionRequest tr = incomingReq.get(transactionId);
		String peer = null;
		if (!tr.getTrans().getPart1().getOwner().equals(
				P2PTrade.getInstance().getPeerName())) {
			peer = tr.getTrans().getPart1().getOwner();
		} else {
			peer = tr.getTrans().getPart2().getOwner();
		}
		PMessage atr = new AuthenticatedMessageWrapper(new ApproveTRequest(
				transactionId));
		try {
			Util.transdferMessageToPeer(atr.getMessageToTransfer(), peer);
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		
		while (ifCommit == null){
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		List<Item> items = null;
		if (P2PTrade.getInstance().getPeerName().equals(
				tr.getTrans().getPart1().getOwner())) {
			items = tr.getTrans().getPart1().getItems();
		} else {
			items = tr.getTrans().getPart2().getItems();
		}
		boolean hasThis = true;
		for (Item item:items){
			if (!ItemRepository.getInstance().hasThis(item.getId())){
				hasThis =  false;
			}
		}
		
		boolean lockMe = LockManager.getInstance().lockMySelf(transactionId);
		if (!lockMe || !hasThis){
			CannotCommit cannot = new CannotCommit(transactionId);
			Util.transdferMessageToPeer(cannot.getMessageToTransfer(), this.ifCommit.fromWhom);
		}else {
			OKCommit ok = new OKCommit(transactionId);
			Util.transdferMessageToPeer(ok.getMessageToTransfer(), this.ifCommit.fromWhom);
		}
		
		while (this.commit == null && this.abortCommit == null){
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		if (this.commit !=null){
			synchronized (mutex) {
				
				this.successReq.put(transactionId, incomingReq.remove(transactionId));
				List<Integer> list = new LinkedList<Integer>();
				for (Item item: items){
					ItemRepository.getInstance().removeById(item.getId());
				}
				LockManager.getInstance().unlockMySelf(transactionId);
				System.out.println(P2PTrade.getInstance().getPeerName()
						+ ":transaction finished!!");
			}
		}else {
			this.transactionRequestAborted(transactionId);
		}
		this.ifCommit = null;
		this.abortCommit = null;
		this.commit = null;
		this.coordinator = null;
		
	}

	public void denyRequest(String transactionId) {
		TransactionRequest tr = this.incomingReq.remove(transactionId);
		String peer = null;
		if (!tr.getTrans().getPart1().getOwner().equals(
				P2PTrade.getInstance().getPeerName())) {
			peer = tr.getTrans().getPart1().getOwner();
		} else {
			peer = tr.getTrans().getPart2().getOwner();
		}

		PMessage dr = new AuthenticatedMessageWrapper(new DenyTRequest(
				transactionId));
		try {
			Util.transdferMessageToPeer(dr.getMessageToTransfer(), peer);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void lockDenied(String transId) {
		if (this.issuedReq.get(transId) != null) {
			this.lockFailedTransId.add(transId);
		}
	}

	public void lockGranted(String transId) {
		System.out.println("lock granted:" + transId);
		Set<String> keys = this.issuedReq.keySet();
		for (String key : keys) {
			System.out.println("issued req:" + key);
		}
		if (this.issuedReq.get(transId) != null) {
			this.lockSuccessTransId.add(transId);
		}

	}

	public Map<String, TransactionRequest> getAbortedReq() {
		return abortedReq;
	}

	public Map<String, TransactionRequest> getDeniedReq() {
		return deniedReq;
	}

	public Map<String, TransactionRequest> getIncomingReq() {
		return incomingReq;
	}

	public Map<String, TransactionRequest> getIssuedReq() {
		return issuedReq;
	}


	public void transactionRequestAborted(String transactionId) {
		if (!this.abortedReq.containsKey(transactionId)) {
			TransactionRequest tr = null;
			if (this.issuedReq.containsKey(transactionId)) {
				tr = issuedReq.remove(transactionId);
			} else if (this.incomingReq.containsKey(transactionId)) {
				tr = incomingReq.remove(transactionId);
			} else {
				return;
			}
			this.abortedReq.put(transactionId, tr);
		}
		if (transactionId.equals(LockManager.getInstance()
				.getCurrentTransactionId())) {
			LockManager.getInstance().unlockMySelf(transactionId);
		}
	}

	public Map<String, TransactionRequest> getSuccessReq() {
		return successReq;
	}

	public void setSuccessReq(Map<String, TransactionRequest> successReq) {
		this.successReq = successReq;
	}
	
	public static class Coordinator implements Runnable{
		String transactionId;
		String peer;
		public Boolean peerCanCommit = null;
		public Boolean meCanCommit = null;
		public Coordinator(String transactionId, String peer){
			this.transactionId = transactionId;
			this.peer = peer;
		}
		public void run() {
			CanCommit canCommit = new CanCommit(transactionId);
			Util.transdferMessageToPeer(canCommit.getMessageToTransfer(), P2PTrade.getInstance().getPeerName());
			Util.transdferMessageToPeer(canCommit.getMessageToTransfer(), peer);
			while (meCanCommit == null || peerCanCommit == null){
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			if (!meCanCommit||!peerCanCommit){
				AbortCommit ab = new AbortCommit(transactionId);
				try {
					Util.transdferMessageToPeer(new AuthenticatedMessageWrapper(ab).getMessageToTransfer(), P2PTrade.getInstance().getPeerName());
					Util.transdferMessageToPeer(new AuthenticatedMessageWrapper(ab).getMessageToTransfer(), peer);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}else {
				Commit com = new Commit(transactionId);
				Util.transdferMessageToPeer(com.getMessageToTransfer(), P2PTrade.getInstance().getPeerName());
				Util.transdferMessageToPeer(com.getMessageToTransfer(), peer);
			}
		}
		
		
	}
	
	public static class CommitInfo{
		String transactionId;
		String fromWhom;
		public CommitInfo(String transactionId, String fromWhom){
			this.transactionId = transactionId;
			this.fromWhom = fromWhom;
		}
		public String getFromWhom() {
			return fromWhom;
		}
		public String getTransactionId() {
			return transactionId;
		}
	}
	
}
